package framework;

/******************************************************************************************************************
 * File:FilterFramework.java
 * Course: 17655
 * Project: Assignment 1
 * Copyright: Copyright (c) 2003 Carnegie Mellon University
 * Versions:
 *	1.0 November 2008 - Initial rewrite of original assignment 1 (ajl).
 *
 * Description:
 *
 * This abstract class defines a skeletal filter framework that defines a filter in terms of the input and output
 * ports. All filters must be defined in terms of this framework - that is, filters must extend this class
 * in order to be considered valid system filters. Filters as standalone threads until the input port no longer
 * has any data - at which point the filter finishes up any work it has to do and then terminates.
 *
 * Parameters:
 *
 * InputReadPort:	This is the filter's input port. Essentially this port is connected to another filter's piped
 *					output steam. All filters connect to other filters by connecting their input ports to other
 *					filter's output ports. This is handled by the Connect() method.
 *
 * OutputWritePort:	This the filter's output port. Essentially the filter's job is to read data from the input port,
 *					perform some operation on the data, then write the transformed data on the output port.
 *
 * FilterFramework:  This is a reference to the filter that is connected to the instance filter's input port. This
 *					reference is to determine when the upstream filter has stopped sending data along the pipe.
 *
 * Internal Methods:
 *
 *	public void Connect( FilterFramework Filter )
 *	public byte ReadFilterInputPort()
 *	public void WriteFilterOutputPort(byte datum)
 *	public boolean EndOfInputStream()
 *
 ******************************************************************************************************************/
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

/**
 * The abstract class that represents a generic filter. 
 */
@SuppressWarnings({"all"})
public abstract class FilterFramework extends Thread {

    // Define filter input and output ports
	/**
	 * The array of input pipes
	 */
    private PipedInputStream InputReadPorts[];
    
    /**
     * The arary of output pipes
     */
    private PipedOutputStream OutputWritePorts[];
    
    /**
     * The following reference to a filter is used because java pipes are able
     * to reliably detect broken pipes on the input port of the filter. 
     * This variable will point to the previous filter in the network and when it dies, 
     * we know that it has closed its output pipe and will send no more data.
     * In this case it is the array of input filters (one per port)
     */
    private FilterFramework InputFilters[];

    /***************************************************************************
     * InnerClass:: EndOfStreamException Purpose: This
     *
     *
     *
     * Arguments: none
     *
     * Returns: none
     *
     * Exceptions: none
     *
     ****************************************************************************/
    class EndOfStreamException extends Exception {

        EndOfStreamException(String s) {
            super(s);
        }
    } // class

    /**
     *
     * Constructor
     *
     * @param numberOfInputs
     *            the number of input pipes for this filter
     * @param numberOfOutputs
     *            the number of output pipes for this filter
     */
    protected FilterFramework(int numberOfInputs, int numberOfOutputs) {
        this.InputReadPorts = new PipedInputStream[numberOfInputs];
        this.OutputWritePorts = new PipedOutputStream[numberOfOutputs];
        this.InputFilters = new FilterFramework[numberOfInputs];

        for (int i = 0; i < numberOfInputs; i++) {
            this.InputReadPorts[i] = new PipedInputStream();
        }

        for (int i = 0; i < numberOfOutputs; i++) {
            this.OutputWritePorts[i] = new PipedOutputStream();
        }
    }

    /***************************************************************************
     * CONCRETE METHOD:: Connect Purpose: This method connects filters to each
     * other. All connections are through the inputport of each filter. That is
     * each filter's inputport is connected to another filter's output port
     * through this method.
     *
     * @param Filter
     *            the filer to connect to
     * @param InputPortIndex
     *            the input port of this filter to connect the pipe
     * @param OutputPortIndex
     *            the output port of the filer to connect to
     */
    public void Connect(FilterFramework Filter, int InputPortIndex,
            int OutputPortIndex) {
        try {
            // Connect this filter's input to the upstream pipe's output stream

            this.InputReadPorts[InputPortIndex].connect(Filter.OutputWritePorts[OutputPortIndex]);
            InputFilters[InputPortIndex] = Filter;

        } // try
        catch (Exception Error) {
            System.out.println("\n" + this.getName()
                    + " FilterFramework error connecting::" + Error);

        } // catch

    } // Connect

    /***************************************************************************
     * CONCRETE METHOD:: ReadFilterInputPort Purpose: This method reads data
     * from the input port one byte at a time.
     *
     * Arguments: void
     *
     * Returns: byte of data read from the input port of the filter.
     *
     * Exceptions: IOException, EndOfStreamException (rethrown)
     *
     ***************************************************************************
     * @param InputPortIndex
     *            the index of the input pipe to read from
     * @return the byte read from the input pipe
     * @throws framework.FilterFramework.EndOfStreamException
     *             when the input is closed
     */
    protected byte ReadFilterInputPort(int InputPortIndex)
            throws EndOfStreamException {
        byte datum = 0;

        /***********************************************************************
         * Since delays are possible on upstream filters, we first wait until
         * there is data available on the input port. We check,... if no data is
         * available on the input port we wait for a quarter of a second and
         * check again. Note there is no timeout enforced here at all and if
         * upstream filters are deadlocked, then this can result in infinite
         * waits in this loop. It is necessary to check to see if we are at the
         * end of stream in the wait loop because it is possible that the
         * upstream filter completes while we are waiting. If this happens and
         * we do not check for the end of stream, then we could wait forever on
         * an upstream pipe that is long gone. Unfortunately Java pipes do not
         * throw exceptions when the input pipe is broken. So what we do here is
         * to see if the upstream filter is alive. if it is, we assume the pipe
         * is still open and sending data. If the filter is not alive, then we
         * assume the end of stream has been reached.
         ***********************************************************************/
        try {
            while (InputReadPorts[InputPortIndex].available() == 0) {
                if (EndOfInputStream(InputPortIndex)) {
                    throw new EndOfStreamException(
                            "End of input stream reached");

                } // if

                sleep(250);

            } // while

        } // try
        catch (EndOfStreamException Error) {
            throw Error;

        } // catch
        catch (Exception Error) {
            System.out.println("\n" + this.getName()
                    + " Error in read port wait loop::" + Error);

        } // catch

        /***********************************************************************
         * If at least one byte of data is available on the input pipe we can
         * read it. We read and write one byte to and from ports.
         ***********************************************************************/
        try {
            datum = (byte) InputReadPorts[InputPortIndex].read();
            return datum;

        } // try
        catch (Exception Error) {
            System.out.println("\n" + this.getName() + " Pipe read error::"
                    + Error);
            return datum;

        } // catch

    } // ReadFilterPort

    /**
     * This methods reads an id (4 bytes) from an input filter
     *
     * @param InputPortIndex
     *            the input filter index to read from
     * @return the id that was read
     * @throws EndOfStreamException
     *             of end of stream is reached
     */
    protected int ReadInt(int InputPortIndex) throws EndOfStreamException {
        int val = 0;

        for (int i = 0; i < 4; i++) {
            byte databyte = ReadFilterInputPort(InputPortIndex); // This is to read the byte from the stream

            val = val | (databyte & 0xFF); // We append the byte on to ID...

            if (i != 4 - 1) // If this is not the last byte, then slide the
            { // previously appended byte to the left by one byte
                val = val << 8; // to make room for the next byte we append to
                // the ID

            } // if

        } // for

        return val;
    }

    /**
     * This methods reads a long (8 bytes) from an input filter
     *
     * @param InputPortIndex
     *            the input filter index to read from
     * @return the measurement that was read
     * @throws EndOfStreamException
     *             of end of stream is reached
     */
    protected long ReadLong(int InputPortIndex) throws EndOfStreamException {
        long val = 0;

        for (int i = 0; i < 8; i++) {
            byte databyte = ReadFilterInputPort(InputPortIndex); // This is
            // where we
            // read the
            // byte from
            // the
            // stream...

            val = val | (databyte & 0xFF); // We append the byte on to ID...

            if (i != 8 - 1) // If this is not the last byte, then slide the
            { // previously appended byte to the left by one byte
                val = val << 8; // to make room for the next byte we append to
                // the ID

            } // if

        } // for

        return val;
    }

    /**
     * Writes an integer to specified output port
     *
     * @param val
     *            the integer to write
     * @param OutputPortIndex
     *            the index of the output port to write to
     */
    protected void writeInt(int val, int OutputPortIndex) {

        this.WriteFilterOutputPort((byte) (val >>> 24), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 16), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 8), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val), OutputPortIndex);
    }

    /**
     * Writes a long to specified output port
     *
     * @param val
     *            the long to write
     * @param OutputPortIndex
     *            the index of the output port to write to
     */
    protected void writeLong(long val, int OutputPortIndex) {
        this.WriteFilterOutputPort((byte) (val >>> 56), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 48), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 40), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 32), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 24), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 16), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val >>> 8), OutputPortIndex);
        this.WriteFilterOutputPort((byte) (val), OutputPortIndex);
    }

    /**
     * CONCRETE METHOD:: WriteFilterOutputPort Purpose: This method writes data
     * to the output port one byte at a time.
     *
     * @param datum
     *            the byte to write to the output pipe
     * @param OutputPortIndex
     *            the output pipe to write the byte to
     */
    protected void WriteFilterOutputPort(byte datum, int OutputPortIndex) {
        try {
            OutputWritePorts[OutputPortIndex].write((int) datum);
            OutputWritePorts[OutputPortIndex].flush();

        } // try
        catch (Exception Error) {
            System.out.println("\n" + this.getName() + " Pipe write error::"
                    + Error);
            //Error.printStackTrace(System.out);

        } // catch

    } // WriteFilterPort

    /**
     * CONCRETE METHOD:: WriteFilterOutputPort Purpose: This method writes data
     * to the output port several byte at a time.
     *
     * @param datum
     *            the byte array to write to the output pipe
     * @param OutputPortIndex
     *            the output pipe to write the byte to
     */
    protected void WriteFilterOutputPort(byte[] datum, int OutputPortIndex) {
        try {
            OutputWritePorts[OutputPortIndex].write(datum);
            OutputWritePorts[OutputPortIndex].flush();

        } // try
        catch (Exception Error) {
            System.out.println("\n" + this.getName() + " Pipe write error::"
                    + Error);

        } // catch

    } // WriteFilterPort

    /***************************************************************************
     * CONCRETE METHOD:: EndOfInputStream Purpose: This method is used within
     * this framework which is why it is private It returns a true when there is
     * no more data to read on the input port of the instance filter. What it
     * really does is to check if the upstream filter is still alive. This is
     * done because Java does not reliably handle broken input pipes and will
     * often continue to read (junk) from a broken input pipe.
     *
     * @param InputPortIndex
     *            the index of the input pipe to check if reached end of input
     *            stream
     * @return true if the input pipe reached end of stream
     */
    private boolean EndOfInputStream(int InputPortIndex) {

        try {
            // thread is not alive -> won't write anything else
            if (!InputFilters[InputPortIndex].isAlive()) {
                // nothing else to read from pipe
                if (this.InputReadPorts[InputPortIndex].available() <= 0) {
                    return true;
                }
            }
        }  catch (IOException e) {
            return true;
        }

        return false;
    } // EndOfInputStream

    /***************************************************************************
     * CONCRETE METHOD:: ClosePorts Purpose: This method is used to close all the
     * input and output ports of the filter. It is important that filters close
     * their ports before the filter thread exits.
     *
     **/
    protected void ClosePorts() {
        try {
            for (PipedInputStream pis : this.InputReadPorts) {
                pis.close();
            }
            for (PipedOutputStream pos : this.OutputWritePorts) {
                pos.close();
            }
        } catch (Exception Error) {
            System.out.println("\n" + this.getName() + " ClosePorts error::"
                    + Error);

        } // catch

    } // ClosePorts

    /***************************************************************************
     * CONCRETE METHOD:: ClosePort Purpose: This method closes the
     * input port of the filter. It is important that filters close
     * their ports before the filter thread exits.
     *
     * @param InputPortIndex the index of the input port to close
     **/
    protected void CloseInputPort(int InputPortIndex) {
        try {

            this.InputReadPorts[InputPortIndex].close();

        } catch (Exception Error) {
            System.out.println("\n" + this.getName() + " ClosePort error::"
                    + Error);

        } // catch

    } // CloseInputPort

    /**
     * Close all the output port
     **/
    protected void CloseOutputPorts() {
        try {

            for (int i= 0; i<OutputWritePorts.length; i++)
                this.OutputWritePorts[i].close();

        } catch (Exception Error) {
            System.out.println("\n" + this.getName() + " CloseOutputPorts error::"
                    + Error);

        } // catch

    } // CloseOutputPorts

    /**
     * Close an output port, by index
     * 
     * @param outputPortIndex the index of the output port to close
     **/
    protected void CloseOutputPort(int outputPortIndex) {
        try {

            this.OutputWritePorts[outputPortIndex].close();

        } catch (Exception Error) {
            System.out.println("\n" + this.getName() + " CloseOutputPort error::"
                    + Error);

        } // catch

    } // CloseOutputPort

    /*
     * CONCRETE METHOD:: run Purpose: This is actually an abstract method
     * defined by Thread. It is called when the thread is started by calling the
     * Thread.start() method. In this case, the run() method should be
     * overridden by the filter programmer using this framework superclass
     */
    public abstract void run();
    // The run method should be overridden by the subordinate class. Please
    // see the example applications provided for more details.
} // FilterFramework class
